package com.ptone.messageCenter.api;

import com.ptone.messageCenter.common.response.Result;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;


@Component
public class ConsumerAPI {


    Logger LOGGER = LoggerFactory.getLogger(getClass());

    @Transactional
    @KafkaListener(topics = {"${consumer.topicName}"}, containerFactory = "ackContainerFactory")
    public Result listen(ConsumerRecord<?, ?> record, Acknowledgment ack) {
        //key - value 即为消息体
        String key = (String) record.key();
        String value = (String) record.value();
        System.out.println(key + "------------" + value);
        LOGGER.info("key:\t" + key + "\t----\tvalue:\t" + value);
        //确认读取消息前需要做的事


        //确认读取消息（类似于事务的commit）
        ack.acknowledge();
        //成功后return
        return Result.end(null);
    }
}
